defmodule SockWorkstation.TTSHelperGen do
  use GenServer

  alias SockWorkstation.{CoopMgrGen, SockCoopClient, CodecNifGen, FrameHelperGen}
  alias SockWorkstation.{CoopHelperSup}
  # 1 MB 最大缓冲区大小
  # @max_buffer_size 1024 * 1024 * 1
  @name __MODULE__
  # 200 毫秒发送间隔
  @send_duration 56
  @chunk_size 1920

  # Client API
  def start_link(session_id) do
    GenServer.start_link(@name, session_id, name: via_tuple(session_id, "tts_helper"))
  end

  def start_tts_pipe() do
    SockWorkstation.TTSHelperGen.TTSALFPipeline.start()
  end

  def get_state(session_id) do
    GenServer.call(via_tuple(session_id, "tts_helper"), :get_state)
  end

  def tts_started(session_id) do
    GenServer.cast(via_tuple(session_id, "tts_helper"), :tts_started)
  end

  def tts_stopped(session_id) do
    GenServer.cast(via_tuple(session_id, "tts_helper"), :tts_stopped)
  end

  def fetch_coop_pid(session_id, pid) do
    GenServer.cast(via_tuple(session_id, "tts_helper"), {:fetch_coop_pid, pid})
  end

  # def rm_coop_pid(session_id) do
  #   GenServer.cast(via_tuple(session_id, "tts_helper"), :rm_coop_pid)
  # end

  def cast_text(session_id, text) do
    GenServer.cast(via_tuple(session_id, "tts_helper"), {:cast_text, text})
  end

  def bcast_rectodb(session_id, db, docid, name, text) do
    entity_msg = "#{db}__#{docid}__#{name}"

    send(
      get_state(session_id).tts_helper_pid,
      {:bcast_tts, {"rec_todb", entity_msg, text}}
    )
  end

  def bcast_recincdb(session_id, db, docid, name, text) do
    entity_msg = "#{db}__#{docid}__#{name}"

    send(
      get_state(session_id).tts_helper_pid,
      {:bcast_tts, {"rec_incdb", entity_msg, text}}
    )
  end

  # def cast_sent_authcode(session_id, text) do
  #   GenServer.cast(via_tuple(session_id, "tts_helper"), {:sent_authcode, text})
  # end

  # def fetch_ttspcm(session_id, pcm_data) do
  #   GenServer.cast(via_tuple(session_id, "tts_helper"), {:fetch_ttspcm, pcm_data})
  # end

  def test_opus(session_id) do
    GenServer.cast(via_tuple(session_id, "tts_helper"), :test_opus)
  end

  def bcast_opusbin(session_id, db, docid, name) do
    GenServer.cast(via_tuple(session_id, "tts_helper"), {:bcast_opusbin, db, docid, name})
  end

  def run_test_opus(session_id) do
    send(get_state(session_id).tts_helper_pid, :run_test_opus)
  end

  def run_test_opus(session_id, db, docid, attachmentid) do
    send(get_state(session_id).tts_helper_pid, {:run_test_opus, {db, docid, attachmentid}})
  end

  # Server Callbacks
  @impl true
  def init(session_id) do
    self_pid = self()
    start_tts_pipe()
    CoopMgrGen.fetch_tts_helper_pid(session_id, self_pid)
    # SockWorkstation.TTSHelperGen.TTSMembrane.start_link(session_id, self_pid)
    # Process.send_after(self_pid, {:run_tts, "今天是星期一，天气晴朗，万里无云"}, 2000)

    {:ok,
     %{
       bin_buffer: <<>>,
       remaining: <<>>,
       session_id: session_id,
       #  user_id: user_id,
       tts_helper_pid: self_pid,
       coop_pid: nil,
       coop_type: :chat,
       #  inter_peer: nil,
       #  coop_retry_count: 0,
       tts_started?: false,
       entity_msg: nil,
       tts_pid: nil,
       frontend_type: :idle,
       backend_type: :tts,
       timer_ref: nil
       #  tts_text: "",
       #  chunk_num: 0
     }}
  end

  @impl true
  def handle_call(:get_state, _from, state) do
    {:reply, state, state}
  end

  @impl true
  def handle_cast({:fetch_coop_pid, coop_pid}, state) when coop_pid == state.coop_pid do
    {:noreply, state}
  end

  def handle_cast({:fetch_coop_pid, coop_pid}, state) do
    {:noreply, %{state | coop_pid: coop_pid}}
  end

  # def handle_cast(:rm_coop_pid, state) do
  #   {:noreply, %{state | coop_pid: nil}}
  # end

  def handle_cast(:tts_started, state) do
    # Process.sleep(120)
    FrameHelperGen.enqueue_pcmbin(state.session_id)
    # FrameHelperGen.start_push(state.session_id)
    # send(state.tts_helper_pid, :tts_pcmchunk)
    {:noreply, %{state | tts_started?: true}}
  end

  def handle_cast(:tts_stopped, state) do
    Process.sleep(120)
    # CoopMgrGen.start_listening(state.session_id)
    # Process.cancel_timer(state.timer_ref)
    cond do
      state.backend_type == :tts ->
        {:noreply, %{state | tts_started?: false, tts_pid: nil, frontend_type: :idle}}

      state.backend_type == :greet_authcode ->
        with [db, docid, name] <- String.split(state.entity_msg, "__") do
          SockWorkstation.CollHelperGen.save_to_couchdb(
            db,
            docid,
            name
          )
        end

        # Process.sleep(120)
        # CoopMgrGen.say_goodbye(state.session_id)

        {:noreply,
         %{state | tts_started?: false, tts_pid: nil, frontend_type: :idle, backend_type: :tts}}

      state.backend_type == :rec_todb ->
        with [db, docid, name] <- String.split(state.entity_msg, "__") do
          SockWorkstation.CollHelperGen.save_to_couchdb(
            db,
            docid,
            name
          )
        end

        # Process.sleep(120)
        # SockCoopClient.send_goodbye(state.coop_pid)

        {:noreply,
         %{
           state
           | tts_started?: false,
             tts_pid: nil,
             frontend_type: :idle,
             entity_msg: nil,
             backend_type: :tts
         }}

      state.backend_type == :rec_incdb ->
        with [db, docid, name] <- String.split(state.entity_msg, "__") do
          SockWorkstation.CollHelperGen.append_to_couchdb(
            db,
            docid,
            name
          )
        end

        # Process.sleep(120)
        # SockCoopClient.send_goodbye(state.coop_pid)

        {:noreply,
         %{
           state
           | tts_started?: false,
             tts_pid: nil,
             frontend_type: :idle,
             entity_msg: nil,
             backend_type: :tts
         }}

      true ->
        {:noreply, %{state | tts_started?: false, tts_pid: nil, frontend_type: :idle}}
    end
  end

  def handle_cast({:cast_text, text}, state) when state.coop_type == :chat do
    # SockWorkstation.SockCoopClient.send_listening(state["self_pid"])
    IO.puts("cast_text_tts")
    # SockWorkstation.SockCoopClient.set_volume(state.coop_pid, 60)
    # SockWorkstation.SockCoopClient.send_speaking(state.coop_pid)

    send(state.tts_helper_pid, {:run_tts, text})
    # {:ok, tts_pid} =
    #   CoopHelperSup.start_tts_client({state.session_id, text})

    # Process.send_after(state.self_pid, :tts_started, 250)

    # state =
    #   state
    #   |> Map.put(:tts_pid, tts_pid)

    {:noreply, state}
  end

  def handle_cast({:cast_text, _text}, state) do
    # SockWorkstation.SockCoopClient.send_listening(state["self_pid"])

    # SockWorkstation.SockCoopClient.set_volume(state.coop_pid, 60)
    # SockWorkstation.SockCoopClient.send_speaking(state.coop_pid)

    # {:ok, tts_pid} =
    #   CoopHelperSup.start_tts_client({state.session_id, text})

    # Process.send_after(state.self_pid, :tts_started, 250)

    # state =
    #   state
    #   |> Map.put(:tts_pid, tts_pid)

    {:noreply, state}
  end

  # def handle_cast({:fetch_ttspcm, pcm_data}, state) do
  #   IO.puts("TTSHelper Received pcm frame length: #{inspect(byte_size(pcm_data))}")
  #   # FrameHelperGen.add_pcm(state.session_id, pcm_data)

  #   bin_buffer = <<state.bin_buffer::binary, pcm_data::binary>>
  #   # {:noreply, state}
  #   {:noreply, %{state | bin_buffer: bin_buffer}}
  # end

  def handle_cast(:test_opus, state) when state.coop_type == :chat do
    # if state.coop_type == :chat do
    SockCoopClient.set_volume(state.coop_pid, 60)
    Process.sleep(150)
    CoopMgrGen.start_speaking(state.session_id)

    # with {:ok, bin} <- File.read("votts.pcm") do
    with {:ok, bin} <-
           SockWorkstation.PanCouchbeamClient.get_stream_attachment(
             "aod_opus_1",
             "song_lvguang",
             "lvguang.pcm"
           ) do
      # with {:ok, bin} <-
      #        SockWorkstation.PanCouchbeamClient.get_stream_attachment(
      #          "rec_opusbin",
      #          "number",
      #          "5"
      #        ),
      #      {:ok, byte_list} <-
      #        SockWorkstation.CollHelperGen.read_frames_with_separator(bin) do
      #   exe_bcast(state.session_id, byte_list)
      send(state.tts_helper_pid, {:runfetch_mscpcm, bin})
      # end
    end

    {:noreply, state}
  end

  def handle_cast(:test_opus, state) do
    {:noreply, state}
  end

  def handle_cast({:bcast_opusbin, db, docid, name}, state) do
    SockCoopClient.set_volume(state.coop_pid, 60)
    Process.sleep(150)
    CoopMgrGen.start_speaking(state.session_id)

    with {:ok, bin} <-
           SockWorkstation.PanCouchbeamClient.get_stream_attachment(
             #  "rec_opusbin",
             #  "number",
             #  "5"
             db,
             docid,
             name
           ),
         {:ok, byte_list} <-
           SockWorkstation.CollHelperGen.read_frames_with_separator(bin) do
      exe_bcast(state.session_id, byte_list)
    end

    {:noreply, state}
  end

  def handle_cast({:request_song, %{sessid: _session_id, song: _song}}, state) do
    {:noreply, state}
  end

  @impl true

  def handle_info({:runfetch_mscpcm, pcm_data}, state) do
    # IO.puts("TTSHelper Received pcm frame length: #{inspect(byte_size(pcm_data))}")
    Process.send_after(state.tts_helper_pid, :tts_pcmchunk, 200)
    # bin_buffer = <<state.bin_buffer::binary, pcm_data::binary>>

    {:noreply, %{state | bin_buffer: pcm_data, tts_started?: true, frontend_type: :run}}
  end

  def handle_info(:run_test_opus, state) when state.coop_type == :chat do
    # with true <- Process.alive?(state.coop_pid) do
    # if state.coop_type == :chat do
    CoopMgrGen.set_volume(state.session_id, 60)
    Process.sleep(120)
    CoopMgrGen.start_speaking(state.session_id)

    # with {:ok, bin} <-File.read("opus.bin") do
    with {:ok, bin} <-
           SockWorkstation.PanCouchbeamClient.get_stream_attachment(
             "aod_opus_1",
             "song_by4",
             "by4"
           ) do
      {:ok, byte_list} = SockWorkstation.CollHelperGen.read_frames_with_separator(bin)
      # FrameHelperGen.push_frames(state.session_id, byte_list)
      exe_bcast(state.session_id, byte_list)
      # Enum.each(byte_list, fn frame ->
      #   # 替换为您的实际发送函数
      #   SockCoopClient.send_binary(state.coop_pid, frame)
      #   Process.sleep(56)
      # end)
      # FrameHelperGen.stop_push(state.session_id)
      # FrameHelperGen.queue_frames(state.session_id, byte_list)
      # FrameHelperGen.start_push(state.session_id)
      # end
    end

    {:noreply, state}
  end

  def handle_info(:run_test_opus, state) do
    # SockWorkstation.SockCoopClient.start_link(state.session_id)
    # uid = CoopMgrGen.get_state(state.session_id).user_id
    # SessionMgr.dyn_coop_client(state.session_id)
    # Process.sleep(1200)

    # send(CoopMgrGen.get_state(state.session_id).coop_mgr_pid, {:run_coop_client, state.session_id})
    # if state.coop_retry_count < 3, do: send(state.tts_helper_pid, :run_test_opus)
    # {:noreply, %{state | coop_retry_count: state.coop_retry_count + 1}}
    {:noreply, state}
  end

  def handle_info({:run_test_opus, {db, docid, attachmentid}}, state)
      when state.coop_type == :chat do
    # if state.coop_type == :chat do
    CoopMgrGen.set_volume(state.session_id, 60)
    Process.sleep(120)
    CoopMgrGen.start_speaking(state.session_id)

    # with {:ok, bin} <-File.read("opus.bin") do
    with {:ok, bin} <-
           SockWorkstation.PanCouchbeamClient.get_stream_attachment(
             db,
             docid,
             attachmentid
           ) do
      {:ok, byte_list} = SockWorkstation.CollHelperGen.read_frames_with_separator(bin)
      # FrameHelperGen.push_frames(state.session_id, byte_list)
      exe_bcast(state.session_id, byte_list)
      # FrameHelperGen.stop_push(state.session_id)
      # FrameHelperGen.queue_frames(state.session_id, byte_list)
      # FrameHelperGen.start_push(state.session_id)
      # Enum.each(byte_list, fn frame ->
      #   # 替换为您的实际发送函数
      #   SockCoopClient.send_binary(state.coop_pid, frame)
      #   Process.sleep(56)
      # end)
      # end
    end

    {:noreply, state}
  end

  def handle_info({:run_test_opus, _}, state) do
    {:noreply, state}
  end

  def handle_info(:run_bcast_authcode, state) when state.coop_type == :chat do
    # with true <- Process.alive?(state.coop_pid) do
    # if state.coop_type == :chat do
    CoopMgrGen.set_volume(state.session_id, 60)
    Process.sleep(120)
    CoopMgrGen.start_speaking(state.session_id)

    # with {:ok, bin} <-File.read("opus.bin") do
    with {:ok, hello_bin} <-
           SockWorkstation.PanCouchbeamClient.get_stream_attachment(
             "rec_opusbin",
             "greet-authcode",
             "greet-authcode-1"
           ),
         {:ok, authcode_bin} <-
           SockWorkstation.PanCouchbeamClient.get_stream_attachment(
             "greet_authcode",
             state.session_id,
             "greet-#{state.session_id}"
           ) do
      {:ok, hello_byte_list} =
        SockWorkstation.CollHelperGen.read_frames_with_separator(hello_bin)

      {:ok, authcode_byte_list} =
        SockWorkstation.CollHelperGen.read_frames_with_separator(authcode_bin)

      byte_list =
        Enum.slice(hello_byte_list, 0..-25//1) ++ Enum.slice(authcode_byte_list, 0..-5//1)

      # Enum.each(byte_list, fn frame ->
      #   # 替换为您的实际发送函数
      #   SockCoopClient.send_binary(state.coop_pid, frame)
      #   Process.sleep(56)
      # end)
      exe_bcast(state.session_id, byte_list)
      # FrameHelperGen.stop_push(state.session_id)
      # FrameHelperGen.queue_frames(state.session_id, byte_list)
      # FrameHelperGen.start_push(state.session_id)
      # end
    end

    {:noreply, state}
  end

  def handle_info(:run_bcast_authcode, state) do
    # SockWorkstation.SockCoopClient.start_link(state.session_id)
    # uid = CoopMgrGen.get_state(state.session_id).user_id
    # CtrlMgrGen.start_peer_coop(state.session_id, uid)
    # SessionMgr.dyn_coop_client(state.session_id)
    # Process.sleep(1200)

    # send(CoopMgrGen.get_state(state.session_id).coop_mgr_pid, {:run_coop_client, state.session_id})
    # if state.coop_retry_count < 3, do: send(state.tts_helper_pid, :run_bcast_authcode)
    # {:noreply, %{state | coop_retry_count: state.coop_retry_count + 1}}
    {:noreply, state}
  end

  def handle_info({:run_tts, text}, state) when state.coop_type == :chat do
    # SockWorkstation.SockCoopClient.send_listening(state["self_pid"])
    # with true <- Process.alive?(state.coop_pid) do
    #  if state.coop_type == :chat do
    IO.puts("run_sent_tts")
    CoopMgrGen.set_volume(state.session_id, 60)

    CoopMgrGen.start_speaking(state.session_id)

    {:ok, tts_pid} =
      CoopHelperSup.super_tts_client({state.session_id, text})

    # Process.send_after(state.self_pid, :tts_started, 250)

    state =
      state
      |> Map.put(:tts_pid, tts_pid)
      |> Map.put(:frontend_type, :run)

    {:noreply, state}
    # end
  end

  def handle_info({:run_tts, _text}, state) do
    {:noreply, state}
  end

  def handle_info({:bcast_tts, {type, entity_msg, text}}, state) do
    if state.coop_type == :chat do
      CoopMgrGen.set_volume(state.session_id, 90)
      Process.sleep(120)
      CoopMgrGen.start_speaking(state.session_id)

      {:ok, tts_pid} =
        CoopHelperSup.super_tts_client({state.session_id, text})

      state =
        state
        |> Map.put(:tts_pid, tts_pid)
        |> Map.put(:backend_type, type)
        |> Map.put(:entity_msg, entity_msg)
        |> Map.put(:frontend_type, :bcast)

      {:noreply, state}
    end
  end

  def handle_info({:rec_tofile, text}, state) when state.coop_type == :chat do
    # if state.coop_type == :chat do
    CoopMgrGen.set_volume(state.session_id, 90)
    Process.sleep(120)
    CoopMgrGen.start_speaking(state.session_id)

    {:ok, tts_pid} =
      CoopHelperSup.super_tts_client({state.session_id, text})

    state =
      state
      |> Map.put(:tts_pid, tts_pid)
      |> Map.put(:backend_type, :tts)
      # |> Map.put(:entity_msg, entity_msg)
      |> Map.put(:frontend_type, :tofile)

    {:noreply, state}
    # end
  end

  def handle_info({:rec_tofile, _text}, state) do
    {:noreply, state}
  end

  def handle_info(:tts_pcmchunk, state) when state.tts_started? do
    schedule_chunk_tts(state.tts_helper_pid)

    cond do
      state.frontend_type == :run ->
        case split_data(state.bin_buffer, @chunk_size) do
          {pcm_chunk, <<>>} ->
            # tts_stopped(state.session_id)
            # schedule_chunk_tts(state.self_pid)
            SockWorkstation.TTSHelperGen.TTSALFPipeline.handle_event(
              {:tts_pcmchunk, state.session_id, state.coop_pid, pcm_chunk}
            )

            {:noreply, %{state | bin_buffer: <<>>, tts_started?: false}}

          {pcm_chunk, remaining_buffers} ->
            SockWorkstation.TTSHelperGen.TTSALFPipeline.handle_event(
              {:tts_pcmchunk, state.session_id, state.coop_pid, pcm_chunk}
            )

            # schedule_chunk_tts(state.self_pid)
            {:noreply, %{state | bin_buffer: remaining_buffers}}
        end

      state.frontend_type == :bcast ->
        case split_data(state.bin_buffer, @chunk_size) do
          {pcm_chunk, <<>>} ->
            # tts_stopped(state.session_id)
            # schedule_chunk_tts(state.self_pid)
            SockWorkstation.TTSHelperGen.TTSALFPipeline.handle_event(
              {:bcast_pcmchunk, state.session_id, state.coop_pid, pcm_chunk}
            )

            {:noreply, %{state | bin_buffer: <<>>}}

          {pcm_chunk, remaining_buffers} ->
            SockWorkstation.TTSHelperGen.TTSALFPipeline.handle_event(
              {:bcast_pcmchunk, state.session_id, state.coop_pid, pcm_chunk}
            )

            # schedule_chunk_tts(state.self_pid)
            {:noreply, %{state | bin_buffer: remaining_buffers}}
        end

      state.frontend_type == :tofile ->
        case split_data(state.bin_buffer, @chunk_size) do
          {_pcm_chunk, <<>>} ->
            # tts_stopped(state.session_id)
            # schedule_chunk_tts(state.self_pid)
            # SockWorkstation.TTSHelperGen.TTSALFPipeline.handle_event(
            #   {:rec_tofile, state.session_id, state.coop_pid, pcm_chunk}
            # )
            #  File.write!("tts-test.pcm", pcm_data, [:append])
            {:noreply, %{state | bin_buffer: <<>>}}

          {pcm_chunk, remaining_buffers} ->
            # SockWorkstation.TTSHelperGen.TTSALFPipeline.handle_event(
            #   {:rec_file, state.session_id, state.coop_pid, pcm_chunk}
            # )
            File.write!("tts-bcast.pcm", pcm_chunk, [:append])
            {:ok, opus_data} = CodecNifGen.encode_pcm(state.session_id, pcm_chunk)
            :ok = SockCoopClient.send_binary(state.coop_pid, opus_data)

            # schedule_chunk_tts(state.self_pid)
            {:noreply, %{state | bin_buffer: remaining_buffers}}
        end

      true ->
        {:noreply, %{state | bin_buffer: <<>>, frontend_type: :idle}}
    end
  end

  def handle_info(:tts_pcmchunk, state) do
    {:noreply, %{state | bin_buffer: <<>>, tts_started?: false}}
  end


  defp split_data(data, size) when byte_size(data) > size do
    <<chunk::binary-size(size), remaining::binary>> = data
    {chunk, remaining}
    # padded_data = data <> :binary.copy(<<0>>, @chunk_size - byte_size(data))
    # string_data = :binary.bin_to_list(data)
    # IO.puts(byte_size(padded_data))
    # {padded_data, <<>>}
  end

  defp split_data(data, size) do
    padded_data = data <> :binary.copy(<<0>>, size - byte_size(data))
    # string_data = :binary.bin_to_list(data)
    # IO.puts(byte_size(padded_data))
    {padded_data, <<>>}
    # <<chunk::binary-size(size), remaining::binary>> = data
    # {chunk, remaining}
  end

  defp schedule_chunk_tts(pid) do
    send(pid, :tts_pcmchunk)
    Process.sleep(@send_duration)
    # Process.send_after(pid, :tts_chunk, @send_duration)
  end

  #   defp parse_entity_msg(entity_msg) do
  #   # 使用 "__" 分割字符串
  #   parts = String.split(entity_msg, "__")

  #   # 模式匹配，确保分割后的部分数量正确
  #   case parts do
  #     [db, docid, name] ->
  #       {:ok, %{"db"=> db, "docid"=> docid, "name"=> name}}

  #     _ ->
  #       {:error, "Invalid format"}
  #   end
  # end

  defp exe_bcast(session_id, byte_lists) do
    # 按顺序执行任务，并等待每一步完成
    task1 = Task.async(fn -> FrameHelperGen.stop_push(session_id) end)
    # 等待 stop_push 完成
    Task.await(task1)

    task2 = Task.async(fn -> FrameHelperGen.queue_frames(session_id, byte_lists) end)
    # 等待 queue_frames 完成
    Task.await(task2)

    # task3 = Task.async(fn -> FrameHelperGen.start_push(session_id) end)
    # 等待 start_push 完成
    # Task.await(task3)

    :ok
  end

  defp via_tuple(session_id, type) do
    {:via, Registry, {SockWorkstation.SessionRegistry, {session_id, type}}}
  end
end

defmodule SockWorkstation.TTSHelperGen.TTSALFPipeline do
  use ALF.DSL
  # use GenServer
  require Logger
  alias SockWorkstation.{SockCoopClient, CodecNifGen, FrameHelperGen}
  # @pcm_chunk_size 1920
  # 预生成静音数据
  # @silence_bytes :binary.copy(<<0>>, @pcm_chunk_size)

  @components [
    # stage(:handle_event),
    # stage(:split_pcm_to_chunks),
    stage(:source_to_processor),
    stage(:sink_to_coop),
    stage(:sink_to_opuscollector),
    stage(:sink_to_framehelper)
  ]


  def source_to_processor({:tts_pcmchunk, session_id, pcm_chunk}, _opts) do
    case CodecNifGen.encode_pcm(session_id, pcm_chunk) do
      {:ok, opus_data} ->
        # SockCoopClient.send_binary(coop_pid, opus_data)
        {:ok, {:opus_data, opus_data}}

      {:error, reason} ->
        {:error, {:encode_error, reason}}
    end
  end

  def sink_to_coop({:opus_data, coop_pid, opus_data}, _opts) do
    # SockWorkstation.SockCoopClient.send_binary(coop_pid, opus_data)
    case SockCoopClient.send_binary(coop_pid, opus_data) do
      :ok ->
        :ok

      {:error, reason} ->
        {:error, reason}
    end
  end

  def sink_to_opuscollector({:opus_data, opus_data}, _opts) do
    # SockWorkstation.SockCoopClient.send_binary(coop_pid, opus_data)
    case SockWorkstation.CollHelperGen.add_frame(opus_data) do
      :ok ->
        :ok

      _ ->
        {:error, "error"}
    end
  end

  def sink_to_framehelper({:opus_data, session_id, opus_data}, _opts) do
    # SockWorkstation.SockCoopClient.send_binary(coop_pid, opus_data)
    case SockWorkstation.FrameHelperGen.append_frame(session_id, opus_data) do
      :ok ->
        :ok

      _ ->
        {:error, "error"}
    end
  end

  def handle_event(event) do
    # 将事件传入管道

    case event do

      {:tts_pcmchunk, session_id, coop_pid, pcm_chunk} ->
        with {:ok, {:opus_data, opus_data}} <-
               source_to_processor({:tts_pcmchunk, session_id, pcm_chunk}, %{}),
             :ok <- sink_to_coop({:opus_data, coop_pid, opus_data}, %{}) do
          :ok
        end

      {:bcast_pcmchunk, session_id, coop_pid, pcm_chunk} ->
        with {:ok, {:opus_data, opus_data}} <-
               source_to_processor({:tts_pcmchunk, session_id, pcm_chunk}, %{}),
             :ok <- sink_to_coop({:opus_data, coop_pid, opus_data}, %{}),
             :ok <- sink_to_opuscollector({:opus_data, opus_data}, %{}) do
          :ok
        end

        # {:tts_pcmchunk, session_id, coop_pid, pcm_chunk} ->
        #   with {:ok, {:opus_data, coop_pid, opus_data}} <-
        #          source_to_processor({:tts_pcmchunk, session_id, coop_pid, pcm_chunk}, %{}),
        #        :ok <- sink_to_coop({:opus_data, coop_pid, opus_data}, %{}),
        #        :ok <- sink_to_opuscollector({:opus_data, coop_pid, opus_data}, %{}) do
        #     :ok
        #   end

        # {:rec_tofile, session_id, coop_pid, pcm_chunk} ->
        #   with {:ok, {:opus_data, coop_pid, opus_data}} <-
        #          source_to_processor({:tts_pcmchunk, session_id, coop_pid, pcm_chunk}, %{}),
        #        #  :ok <- sink_to_coop({:opus_data, coop_pid, opus_data}, %{}),
        #        :ok <- sink_to_opuscollector({:opus_data, coop_pid, opus_data}, %{}) do
        #     :ok
        #   end
    end
  end
end

defmodule SockWorkstation.TTSHelperGen.CustomSource do
  use Membrane.Source

  def_options(
    session_id: [type: :string],
    coop_pid: [type: :pid]
  )

  def_output_pad(:output,
    flow_control: :manual,
    demand_unit: :buffers,
    accepted_format: _any
  )

  # def_input_pad :input, caps: :any

  #  def start_link(options) do
  #   GenServer.start_link(__MODULE__, options, name: via_tuple(options.session_id, "membrane_source"))
  # end

  def handle_init(_ctx, options) do
    self_pid = self()
    SockWorkstation.CoopMgrGen.fetch_membrane_source_pid(options.session_id, self_pid)

    state = %{
      self_pid: self_pid,
      session_id: options.session_id,
      coop_pid: options.coop_pid,
      buffers: []
    }

    {[], state}
  end

  @impl true
  def handle_demand(:output, size, :buffers, _ctx, state) do
    # 从缓冲区中取出数据并发送
    {buffers_to_send, remaining_buffers} = Enum.split(state.buffers, size)

    # 创建 Membrane.Buffer 并发送
    buffers =
      Enum.map(buffers_to_send, fn buffer_data ->
        %Membrane.Buffer{payload: buffer_data}
      end)

    # 发送缓冲区数据到下一个组件
    actions =
      Enum.map(buffers, fn buffer ->
        {:buffer, :output, buffer}
      end)

    {actions, %{state | buffers: remaining_buffers}}
  end

  def handle_info({:push_pcm_data, chunk}, _ctx, state) do
    # 将 PCM 数据添加到缓冲区
    buffer = %Membrane.Buffer{payload: chunk}
    {[], %{state | buffers: state.buffers ++ [buffer]}}
  end

  #  def push_pcm_data(session_id, chunk) do
  #     GenServer.call(via_tuple(session_id, "membrane_source"), {:push_pcm_data, chunk})
  #   end

  # defp via_tuple(session_id, type) do
  #   {:via, Registry, {SockWorkstation.SessionRegistry, {session_id, type}}}
  # end
end

defmodule SockWorkstation.TTSHelperGen.TTSSink do
  use Membrane.Sink

  def_options(
    session_id: [type: :string],
    coop_pid: [type: :pid]
  )

  def_input_pad(:input, accepted_format: _any)
  # def start_link([session_id, coop_pid]) do
  #   GenServer.start_link(__MODULE__, [session_id, coop_pid])
  # end

  @impl true
  def handle_init(_ctx, options) do
    state = %{session_id: options.session_id, coop_pid: options.coop_pid, buffer: <<>>}

    {[], state}
  end

  # @impl true
  def handle_buffer(_ctx, buffer, state) do
    IO.puts("Received buffer: #{inspect(byte_size(buffer.payload))} #{inspect(state)}")
    {[], state}
  end
end

defmodule SockWorkstation.TTSHelperGen.TTSMembrane do
  use Membrane.Pipeline

  def start_link(session_id, coop_pid) do
    # 创建管道实例并启动
    options = %{session_id: session_id, coop_pid: coop_pid}
    Membrane.Pipeline.start_link(__MODULE__, options, name: via_tuple(session_id, "tts_membrane"))
  end

  def handle_init(_ctx, options) do
    # 定义管道结构
    # child(:source, %Membrane.File.Source{location: "./tts-test.pcm"})
    structure =
      child(:source, %SockWorkstation.TTSHelperGen.CustomSource{
        session_id: options.session_id,
        coop_pid: options.coop_pid
      })
      |> via_out(:output)
      |> child(:encoder, %Membrane.Opus.Encoder{
        input_stream_format: %Membrane.RawAudio{
          channels: 1,
          sample_format: :s16le,
          sample_rate: 16_000
        }
      })
      |> child(:sink, %SockWorkstation.TTSHelperGen.TTSSink{
        session_id: options.session_id,
        coop_pid: options.coop_pid
      })

    # links = [
    #   link(:source) |> to(:sink)
    # ]

    # |> via_in(:input)  # 明确指定输入 pad 名称

    {[spec: structure], options}
  end

  def send_to_processor(chunk) do
    # 获取 CustomSource 的 PID
    # source_pid = Process.whereis(:source)
    # source_pid = Membrane.Pipeline.get_child(self(), :source)
    send(
      SockWorkstation.CoopMgrGen.get_state("c87e77dd").membrane_source_pid,
      {:push_pcm_data, chunk}
    )

    # case GenServer.call(source_pid, {:push_pcm_data, chunk}) do
    #   # case SockWorkstation.TTSHelperGen.CustomSource.push_pcm_data(session_id, {:push_pcm_data, chunk}) do
    #   :ok ->
    #     # 等待编码后的 Opus 数据
    #     receive do
    #       {:new_buffer, opus_data} ->
    #         {:ok, {:opus_data, opus_data}}
    #     end

    #   {:error, reason} ->
    #     {:error, {:encode_error, reason}}
    # end
  end

  defp via_tuple(session_id, type) do
    {:via, Registry, {SockWorkstation.SessionRegistry, {session_id, type}}}
  end
end

 # def handle_info(:tts_pcm_chunk, state) when state.tts_started? do
  #   schedule_chunk_tts(state.tts_helper_pid)

  #   case split_data(state.bin_buffer, @chunk_size) do
  #     {pcm_chunk, <<>>} ->
  #       # tts_stopped(state.session_id)
  #       # schedule_chunk_tts(state.self_pid)
  #       {:ok, opus_data} = CodecNifGen.encode_pcm(state.session_id, pcm_chunk)
  #       FrameHelperGen.append_frame(state.session_id, opus_data)
  #       {:noreply, %{state | bin_buffer: <<>>, tts_started?: false}}

  #     {pcm_chunk, remaining_buffers} ->
  #       {:ok, opus_data} = CodecNifGen.encode_pcm(state.session_id, pcm_chunk)
  #       FrameHelperGen.append_frame(state.session_id, opus_data)

  #       # schedule_chunk_tts(state.self_pid)
  #       {:noreply, %{state | bin_buffer: remaining_buffers}}
  #   end
  # end

  # def handle_info(:tts_pcm_chunk, state) do
  #   {:noreply, %{state | bin_buffer: <<>>, tts_started?: false}}
  # end

  # def handle_info(:tts_pcmqueue, state) when state.tts_started? do
  #   cond do
  #     state.frontend_type == :run ->
  #       case split_data(state.bin_buffer, @chunk_size) do
  #         {pcm_chunk, <<>>} ->
  #           # tts_stopped(state.session_id)
  #           # schedule_chunk_tts(state.self_pid)
  #           # with {:ok, opus_data} <- CodecNifGen.encode_pcm(state.session_id, pcm_chunk) do
  #             IO.puts("pcm_chunk: #{byte_size(pcm_chunk)}")
  #             FrameHelperGen.add_pcm(state.session_id, pcm_chunk)
  #           # end

  #           {:noreply, %{state | bin_buffer: <<>>, tts_started?: false}}

  #         {pcm_chunk, remaining_buffers} ->
  #           # schedule_chunk_tts(state.self_pid)
  #           # with {:ok, opus_data} <- CodecNifGen.encode_pcm(state.session_id, pcm_chunk) do
  #             IO.puts("pcm_chunk: #{byte_size(pcm_chunk)}")
  #             FrameHelperGen.add_pcm(state.session_id, pcm_chunk)
  #           # end

  #           {:noreply, %{state | bin_buffer: remaining_buffers}}
  #       end

  #     true ->
  #       {:noreply, %{state | bin_buffer: <<>>, frontend_type: :idle}}
  #   end
  # end

  # def handle_info(:tts_pcmqueue, state) do
  #   {:noreply, %{state | bin_buffer: <<>>, tts_started?: false}}
  # end

  # def handle_info({:inter_com, {fr_sessid, to_sessid}}, state) do
  #   # SockWorkstation.SockCoopClient.start_link(state.session_id)
  #   # uid=CoopMgrGen.get_state(state.session_id).user_id
  #   # SessionMgr.dyn_coop_client(state.session_id)
  #   # CoopHelperSup.super_coop_client(state.session_id)
  #   # Process.sleep(1200)
  #   # if state.coop_retry_count < 3, do: send(state.tts_helper_pid, {:run_tts, text})
  #   # {:noreply, %{state | coop_retry_count: state.coop_retry_count + 1}}
  #   {:noreply, %{state | coop_type: :inter_com, inter_peer: {fr_sessid, to_sessid}}}
  # end

  # def handle_info(:inter_com, state) do
  #   # with {fr_sessid, to_sessid} <- state.inter_peer do
  #   #   FrameHelperGen.stop_push(to_sessid)
  #   #   FrameHelperGen.stop_push(fr_sessid)
  #   # end

  #   {:noreply, %{state | coop_type: :chat, inter_peer: nil}}
  # end

  # @impl true
  # def terminate(_reason, state) do
  #   CoopMgrGen.rm_tts_helper_pid(state.session_id)
  #   :ok
  # end


    # {:tts_pcm, session_id, pcm_data} ->
      #   case split_pcm_to_chunks({:tts_pcm, pcm_data}, %{}) do
      #     chunks when is_list(chunks) ->
      #       # IO.puts("chunks: #{inspect(chunks)}")
      #       chunks
      #         |> Enum.map(fn {:tts_pcmchunk, chunk} -> chunk end)  # 提取二进制数据
      #         |> Enum.map(fn chunk -> {:tts_pcmchunk, session_id, chunk} end)
      #         |> Enum.map(fn chunk -> source_to_processor(chunk, %{}) end)
      #         |> IO.inspect()

      #       # 2. 流式处理每个chunk
      #       # chunks
      #       # |> Enum.reduce_while(:ok, fn pcm_chunk, _acc ->
      #       #   # 3. 编码处理
      #       #   case source_to_processor({:tts_pcmchunk, session_id, pcm_chunk}, %{}) do
      #       #     {:ok, {:opus_data, opus_data}} ->
      #       #       # 4. 下沉处理
      #       #       case sink_to_framehelper({:opus_data, session_id, opus_data}, %{}) do
      #       #         :ok ->
      #       #           # 继续处理下一个chunk
      #       #           {:cont, :ok}

      #       #         error ->
      #       #           Logger.error("Sink failed: #{inspect(error)}")
      #       #           # 停止并返回错误
      #       #           {:halt, error}
      #       #       end

      #       #     {:error, reason} ->
      #       #       Logger.error("Encode failed: #{inspect(reason)}")
      #       #       {:halt, {:error, :encode_failed}}
      #       #   end
      #       # end)
      #       # |> case do
      #       #   :ok ->
      #       #     Logger.debug("Pipeline processed #{length(chunks)} chunks successfully")
      #       #     # 可选：触发后续处理
      #       #     # FrameHelperGen.start_push(session_id)
      #       #     :ok

      #       #   error ->
      #       #     error
      #       # end

      #     error ->
      #       Logger.error("Chunking failed: #{inspect(error)}")
      #       {:error, :chunking_failed}
      #   end

      # with {:tts_pcmchunk, pcm_chunk} <-
      #        split_pcm_to_chunks({:tts_pcm, pcm_data}, %{}) do

      #     #  {:ok, {:opus_data, opus_data}} <-
      #       #  source_to_processor({:tts_pcmchunk, session_id, pcm_chunk}, %{}),
      #     #  :ok <- sink_to_framehelper({:opus_data, session_id, opus_data}, %{}) do
      #   IO.puts("Pipeline executed successfully: #{inspect(event)}")
      #   # FrameHelperGen.start_push(session_id)
      #   :ok
      # end

# def split_pcm_to_chunks({:tts_pcm, pcm_data}, _opts) do
  #   total_size = byte_size(pcm_data)
  #   chunk_count = div(total_size, @pcm_chunk_size)
  #   remainder = rem(total_size, @pcm_chunk_size)

  #   # 处理完整块
  #   {complete_chunks, rest} =
  #     if chunk_count > 0 do
  #       <<complete::binary-size(chunk_count * @pcm_chunk_size), rest::binary>> = pcm_data
  #       {String.codepoints(complete), rest}
  #     else
  #       {[], pcm_data}
  #     end

  #   # 处理剩余部分（填充静音）
  #   padded_chunks =
  #     if remainder > 0 do
  #       padding_size = @pcm_chunk_size - remainder
  #       padding = binary_part(@silence_bytes, 0, padding_size)
  #       padded = <<rest::binary, padding::binary>>
  #       String.codepoints(padded)
  #     else
  #       []
  #     end

  #   # 合并所有块
  #   (complete_chunks ++ padded_chunks)
  #   |> Enum.chunk_every(@pcm_chunk_size)
  #   |> Enum.map(&Enum.join/1)
  #   |> Enum.map(fn chunk ->
  #     {:tts_pcmchunk, chunk}
  #   end)
  # end

  # def split_pcm_to_chunks({:tts_pcm, pcm_data}, _opts) do
  #   total_size = byte_size(pcm_data)
  #   chunk_count = div(total_size, @pcm_chunk_size)
  #   remainder = rem(total_size, @pcm_chunk_size)

  #   # 处理完整块
  #   # 处理完整块
  #   chunks =
  #     if chunk_count > 0 do
  #       <<complete_chunks::binary-size(chunk_count * @pcm_chunk_size), _rest::binary>> = pcm_data
  #       complete_chunks
  #     else
  #       <<>>
  #     end

  #   # 处理剩余部分（填充静音）
  #   padded_rest =
  #     if remainder > 0 do
  #        @silence_bytes
  #       # <<rest::binary-size(remainder), padding::binary>> = {rest, @silence_bytes}
  #       # <<rest::binary, binary_part(@silence_bytes, 0, @pcm_chunk_size - remainder)>>
  #     else
  #       <<>>
  #     end

  #   # 合并所有块
  #   all_chunks =
  #     if byte_size(chunks) > 0 and byte_size(padded_rest) > 0 do
  #       <<chunks::binary, padded_rest::binary>>
  #     else
  #       chunks <> padded_rest
  #     end

  #   # 转换为处理事件
  #   all_chunks
  #   |> String.codepoints()
  #   |> Enum.chunk_every(@pcm_chunk_size)
  #   |> Enum.map(&Enum.join/1)
  #   |> Enum.map(fn chunk ->
  #     {:tts_pcmchunk, chunk}
  #   end)

  #   # if chunk_count > 0 do
  #   #   # 提取完整块，直接丢弃尾部不足部分
  #   #   <<chunks::binary-size(chunk_count * @pcm_chunk_size), _rest::binary>> = pcm_data

  #   #   # 转换为字符列表后分块（确保精确分割）
  #   #   chunks
  #   #   |> String.codepoints()
  #   #   |> Enum.chunk_every(@pcm_chunk_size)
  #   #   |> Enum.map(&Enum.join/1)
  #   #   |> Enum.map(fn chunk ->
  #   #     {:tts_pcmchunk, chunk}
  #   #   end)
  #   # else
  #   #   # 不足一个完整块，全部丢弃
  #   #   Logger.debug("Discarded PCM data: #{total_size} bytes (less than #{@pcm_chunk_size})")
  #   #   []
  #   # end
  # end

  # def source_to_processor({:tts_pcmchunk, session_id, coop_pid, pcm_chunk}, _opts) do
  #   case CodecNifGen.encode_pcm(session_id, pcm_chunk) do
  #     {:ok, opus_data} ->
  #       # SockCoopClient.send_binary(coop_pid, opus_data)
  #       {:ok, {:opus_data, coop_pid, opus_data}}

  #     {:error, reason} ->
  #       {:error, {:encode_error, reason}}
  #   end
  # end

# SockWorkstation.SockCoopClient.start_link(state.session_id)
# uid=CoopMgrGen.get_state(state.session_id).user_id
# SessionMgr.dyn_coop_client(state.session_id)
# CoopHelperSup.super_coop_client(state.session_id)
# Process.sleep(1200)
# if state.coop_retry_count < 3, do: send(state.tts_helper_pid, {:run_tts, text})
# {:noreply, %{state | coop_retry_count: state.coop_retry_count + 1}}

# def handle_cast({:sent_authcode, text}, state) do
#   # SockWorkstation.SockCoopClient.send_listening(state["self_pid"])
#   IO.puts("cast_sent_authcode")
#   # SockWorkstation.SockCoopClient.set_volume(state.coop_pid, 60)
#   # SockWorkstation.SockCoopClient.send_speaking(state.coop_pid)
#   Process.sleep(500)
#   send(state.self_pid, {:run_tts, text})
#   Process.sleep(15000)
#   SockCoopClient.send_goodbye(state.coop_pid)
#   # {:ok, tts_pid} =
#   #   CoopHelperSup.start_tts_client({state.session_id, text})

#   # Process.send_after(state.self_pid, :tts_started, 250)

#   # state =
#   #   state
#   #   |> Map.put(:tts_pid, tts_pid)

#   {:noreply, state}
# end

# def handle_info({:bcast_tts, {type ,text}}, state) do
#   SockCoopClient.set_volume(state.coop_pid, 90)
#   Process.sleep(120)
#   SockCoopClient.send_speaking(state.coop_pid)

#   {:ok, tts_pid} =
#     CoopHelperSup.start_tts_client({state.session_id, text})

#   state =
#     state
#     |> Map.put(:tts_pid, tts_pid)
#     |> Map.put(:backend_type, type)
#     |> Map.put(:frontend_type, "broadcast")

#   {:noreply, state}
# end

# def handle_info(:authcode_chunk, state) when state.tts_started? do
#   schedule_chunk_tts(state.tts_helper_pid)

#   case split_data(state.bin_buffer, @chunk_size) do
#     {pcm_chunk, <<>>} ->
#       # tts_stopped(state.session_id)
#       # schedule_chunk_tts(state.self_pid)
#       SockWorkstation.TTSHelperGen.TTSALFPipeline.handle_event(
#         {:authcode_pcmchunk, state.session_id, state.coop_pid, pcm_chunk}
#       )

#       {:noreply, %{state | bin_buffer: <<>>}}

#     {pcm_chunk, remaining_buffers} ->
#       SockWorkstation.TTSHelperGen.TTSALFPipeline.handle_event(
#         {:authcode_pcmchunk, state.session_id, state.coop_pid, pcm_chunk}
#       )

#       # schedule_chunk_tts(state.self_pid)
#       {:noreply, %{state | bin_buffer: remaining_buffers}}
#   end
# end

# def handle_info(:authcode_chunk, state) do
#   {:noreply, %{state | bin_buffer: <<>>}}
# end

# defmodule SimplePipeline do
#   use Membrane.Pipeline

#     @impl true
#   def start_link do
#     # 启动管道
#     Membrane.Pipeline.start_link(__MODULE__, %{})
#   end

#   @impl true
#   def handle_init(_ctx, _options) do
#     # 定义管道结构
#     children = %{
#       source: %Membrane.File.Source{location: "./tts-test.pcm"},
#       # sink: %Membrane.Element.PortAudio.Sink{}
#     }

#     links = [
#       link(:source) |> to(:sink)
#     ]

#     {{:ok, spec: %Membrane.ChildrenSpec{children: children, links: links}}, %{}}
#   end

# end

# 启动管道
# {:ok, pipeline} = SimplePipeline.start_link()

# case call(event) do
#   # {:ok, {:opus_data, session_id, coop_pid, opus_data}}->
#     # {:opus_data, session_id, coop_pid, opus_data}
#   {:ok, result} ->
#     IO.puts("Pipeline executed successfully: #{inspect(result)}")
#     {:ok, result}

#   {:error, reason} ->
#     log_error(reason)
#     {:error, reason}

#   unknown_result ->
#     log_unknown_event(unknown_result)
#     {:error, :unknown_result}
# end

# defp log_unknown_event(event) do
#   # IO.puts("Unknown event: #{inspect(event)}")
#   IO.puts("event: #{inspect(event, pretty: true, width: 80)}")
# end

# defp log_error(error) do
#   IO.puts("TTSALFPipeline error: #{inspect(error)}")
# end

# defp via_tuple(session_id, type) do
#   {:via, Registry, {SockWorkstation.SessionRegistry, {session_id, type}}}
# end

# File.write!("tts-test.pcm", pcm_data, [:append])
# SockWorkstation.SockCoopClient.send_binary(state.coop_pid, opus)
# case SockWorkstation.CodecNifGen.encode_pcm(state.session_id, pcm_data) do
# {:ok, opus_data} ->
# SockWorkstation.SockCoopClient.send_binary(state.coop_pid, opus_data)

# {:error, reason} ->
#   {:error, {:encode_error, reason}}
# end
# SockWorkstation.TTSHelperGen.TTSALFPipeline.handle_event(
#   {:pcm_data, state.session_id, state.coop_pid, pcm_data}
# )
# def split_buffers(binary, size) when is_binary(binary) and is_integer(size) and size > 0 do
#   case byte_size(binary) do
#     # 如果二进制数据小于等于指定大小，全部发送
#     len when len <= size ->
#       {binary, <<>>}

#     # 如果二进制数据大于指定大小，分离出指定大小的数据
#     _ ->
#       <<buffers_to_send::binary-size(size), remaining_buffers::binary>> = binary
#       {buffers_to_send, remaining_buffers}
#   end
# end

# def chunk_pcm_data(pcm_data, chunk_size) do
#   pcm_data
#   |> Stream.unfold(fn data ->
#     if byte_size(data) >= chunk_size do
#       <<chunk::binary-size(chunk_size), rest::binary>> = data
#       {chunk, rest}
#     else
#       nil
#     end
#   end)
#   |> Enum.to_list()
# end

# def buffer_pcm_data({:pcm_data, session_id, coop_pid, pcm_data}, _opts) do
#   # 将 PCM 数据分块
#   chunks=chunk_pcm_data(pcm_data, @chunk_size)

#   # 将每个块包装为事件
#   events = Enum.map(chunks, fn chunk -> {:pcm_chunk, session_id, coop_pid, chunk} end)

#   # 返回事件列表，管道会逐个处理
#   {:ok, events}
# end
# def chunk_pcm_data(pcm_data, chunk_size \\ @chunk_size) do
#   do_chunk_pcm_data(pcm_data, chunk_size, [])
# end

# defp do_chunk_pcm_data(pcm_data, chunk_size,acc) do
#   case byte_size(pcm_data) do
#     # 如果数据小于块大小，丢弃剩余数据并返回结果
#     len when len < chunk_size ->
#       Enum.reverse(acc)

#     # 如果数据大于等于块大小，分离出指定大小的数据并继续处理
#     _ ->
#       <<chunk::binary-size(chunk_size), remaining_data::binary>> = pcm_data

#       # 延时 60 毫秒
#       Process.sleep(@delay_ms)

#       # 将当前块加入结果列表，并递归处理剩余数据
#       do_chunk_pcm_data(remaining_data, chunk_size, [chunk | acc])
#   end
# case byte_size(pcm_data) do
#   # 如果数据小于块大小，丢弃剩余数据
#   len when len < chunk_size ->
#     :ok

#   # 如果数据大于等于块大小，分离出指定大小的数据并处理
#   _ ->
#     <<chunk::binary-size(chunk_size), remaining_data::binary>> = pcm_data

#     # 处理当前块（例如发送到网络或写入文件）
#     # process_chunk(chunk)

#     # 延时 60 毫秒
#     Process.sleep(@delay_ms)

#     # 递归处理剩余数据
#     do_chunk_pcm_data(remaining_data, chunk_size)
# end

# def chunk_pcm_data(pcm_data, chunk_size) do
#   pcm_data
#   |> Stream.unfold(fn data ->
#     if byte_size(data) >= chunk_size do
#       <<chunk::binary-size(chunk_size), rest::binary>> = data
#       {chunk, rest}
#     else
#       nil
#     end
#   end)
#   |> Enum.to_list()
# end
# def split_buffers(buffer, chunk_size) do
#   if byte_size(buffer) < chunk_size do
#     {[], buffer}
#   else
#     <<send_chunk::binary-size(chunk_size), remaining::binary>> = buffer
#     {send_buffers, remaining} = split_buffers(remaining, chunk_size)
#     {[send_chunk | send_buffers], remaining}
#   end
# end

# 将 PCM 数据推送到 CustomSource
# case Membrane.Element.send_info(source_pid, {:push_pcm_data, chunk}) do
#   :ok ->
#     # 等待编码后的 Opus 数据
#     receive do
#       {:new_buffer, opus_data} ->
#         {:ok, {:opus_data, session_id, coop_pid, opus_data}}
#     end

#   {:error, reason} ->
#     {:error, {:encode_error, reason}}
# end

# defmodule SockWorkstation.TTSHelperGen.TTSMembrane do
#   use Membrane.Pipeline

#   def handle_init(_ctx, _options) do
#     # 定义管道结构
#     structure =
#       child(:source, CustomSource)  # 自定义源组件，用于接收 PCM 数据
#       |> child(:encoder, %Membrane.Opus.Encoder{
#         input_stream_format: %Membrane.RawAudio{
#           channels: 1,
#           sample_format: :s16le,
#           sample_rate: 16_000
#         }
#       })
#       |> child(:sink, TTSSink)

#     {[spec: structure], %{}}
#   end

# defmodule TTSSink do
#   use Membrane.Sink

#   # @impl true
#   def handle_init(_ctx, _options) do
#     {:ok, %{}}
#   end

#   # @impl true
#   def handle_buffer(_ctx, buffer, state) do
#     IO.puts("Received buffer: #{inspect(buffer)}")
#     {:ok, state}
#   end
# end

#   defmodule CustomSource do
#     use Membrane.Source

#     def_options session_id: [type: :string],
#                 coop_pid: [type: :pid]

#     def handle_init(_ctx, options) do
#       state = %{session_id: options.session_id, coop_pid: options.coop_pid, buffers: []}
#       {[], state}
#     end

#     def handle_demand(:output, size, _ctx, state) do
#       # 从缓冲区中取出数据并发送
#       {buffers_to_send, remaining_buffers} = Enum.split(state.buffers, size)

#       # 创建 Membrane.Buffer 并发送
#       buffers = Enum.map(buffers_to_send, fn buffer_data ->
#         %Membrane.Buffer{payload: buffer_data}
#       end)

#       # 发送缓冲区数据到下一个组件
#       actions = Enum.map(buffers, fn buffer ->
#         {:buffer, :output, buffer}
#       end)

#       {actions, %{state | buffers: remaining_buffers}}
#     end

#     def handle_info({:push_pcm_data, chunk}, _ctx, state) do
#       # 将 PCM 数据添加到缓冲区
#       buffer = %Membrane.Buffer{payload: chunk}
#       {[], %{state | buffers: state.buffers ++ [buffer]}}
#     end
#   end

#   @doc """
#   将 PCM 数据推送到管道中。
#   """
#   def send_to_processor({:pcm_chunk, session_id, coop_pid, chunk}, _opts) do
#     # 将 PCM 数据推送到自定义源组件
#     case GenServer.call(:source, {:push_pcm_data, chunk}) do
#       :ok ->
#         # 等待编码后的 Opus 数据
#         receive do
#           {:new_buffer, opus_data} ->
#             {:ok, {:opus_data, session_id, coop_pid, opus_data}}
#         end

#       {:error, reason} ->
# {:error, {:encode_error, reason}}
# end
# end
# end

# defmodule SockWorkstation.TTSHelperGen.TTSMembrane do
#   use Membrane.Pipeline
# _

# def_options session_id: [type: :string],
# coop_pid: [type: :pid]

# def handle_init(_ctx, options) do
# state = %{session_id: options.session_id, coop_pid: options.coop_pid, buffers: []}
# {[], state}
# end

# def handle_demand(:output, size, _unit, _ctx, state) do
# # 从缓冲区中取出数据并发送
# {buffers_to_send, remaining_buffers} = Enum.split(state.buffers, size)
# actions = Enum.map(buffers_to_send, &buffer(:output, &1))
# {actions, %{state | buffers: remaining_buffers}}
# end

# def handle_info({:push_pcm_data, chunk}, _ctx, state) do
# # 将 PCM 数据添加到缓冲区
# buffer = %Membrane.Buffer{payload: chunk}
# {[], %{state | buffers: state.buffers ++ [buffer]}}
# end

# def send_to_processor({:pcm_chunk, session_id, coop_pid, chunk}, _opts) do
# # 将 PCM 数据推送到自定义源组件
# case GenServer.call(:source, {:push_pcm_data, chunk}) do
# :ok ->
# # 等待编码后的 Opus 数据
# receive do
# {:new_buffer, opus_data} ->
# {:ok, {:opus_data, session_id, coop_pid, opus_data}}
# end

# {:error, reason} ->
# {:error, {:encode_error, reason}}
# end
# end
# impl true
# def handle_init(_ctx, _options) do
#   # 定义管道结构
#   structure =
#     child(:source, %Membrane.RemoteStream{})  # 使用 RemoteStream 作为源
# |> child(:encoder, %Membrane.Opus.Encoder{
#   input_stream_format: %Membrane.RawAudio{
#     channels: 1,
#     sample_format: :s16le,
#     sample_rate: 16_000
#   }
# })
#     |> child(:sink, %Membrane.Stream.Sink{
#       location: "output.opus"
#     })

#   {[spec: structure], %{}}
# end

# def send_to_processor({:pcm_chunk, session_id, coop_pid, chunk}, _opts) do
#   # 将 PCM 数据发送到 RemoteStream
#   case Membrane.RemoteStream.push(:source, chunk) do
#     :ok ->
#       # 等待编码后的 Opus 数据
#       receive do
#         {:new_buffer, opus_data} ->
#           {:ok, {:opus_data, session_id, coop_pid, opus_data}}
#       end

#     {:error, reason} ->
#       {:error, {:encode_error, reason}}
#   end
# end
# def start_link() do
#   spec = [
#     %Membrane.Stream.Source{name: :pcm_source},
#     Membrane.Opus.Encoder,
#     %Membrane.Stream.Sink{name: :opus_sink}
#   ]

#   Pipeline.start_link(__MODULE__, spec)
# end

# def send_to_processor({:pcm_chunk, session_id, coop_pid, chunk}, _opts) do
#   # 将 PCM 数据发送到 Membrane 管道进行编码
#   case Membrane.Stream.Source.push(:pcm_source, chunk) do
#     :ok ->
#       # 等待编码后的 Opus 数据
#       receive do
#         {:new_buffer, opus_data} ->
#           {:ok, {:opus_data, session_id, coop_pid, opus_data}}
#       end

#     {:error, reason} ->
#       {:error, {:encode_error, reason}}
#   end
# end
# end
# defp chunk_pcm_data(pcm_data, chunk_size) do
#   pcm_data
#   |> :binary.bin_to_list()
#   |> Enum.chunk_every(chunk_size)
#   |> Enum.map(&:binary.list_to_bin/1)
# end

# defp chunk_pcm_data(pcm_data, chunk_size) do
#   chunk_binary(pcm_data, chunk_size, [])
# end

# defp chunk_binary(<<chunk::binary-size(chunk_size), rest::binary>>, chunk_size, acc) do
#   chunk_binary(rest, chunk_size, [chunk | acc])
# end

# defp chunk_binary(rest, _chunk_size, acc) do
#   # 处理剩余的数据（不足一个 chunk_size 的部分）
#   if byte_size(rest) > 0 do
#     Enum.reverse([rest | acc])
#   else
#     Enum.reverse(acc)
#   end
# end

# def send_to_processor(session_id, pcm_data) do
#   case CodecNifGen.encode_pcm(session_id, pcm_data) do
#     {:ok, opus_data} ->
#       {:ok, opus_data}

#     {:error, reason} ->
#       IO.puts("Error encoding PCM data: #{reason}")
#       {:error, reason}
#   end
# end

# def send_to_coop(coop_pid, opus_data) do
#   # SockWorkstation.SockASRClient.send_asr_task(asr_pid, pcm_data)
#   case SockCoopClient.send_binary(coop_pid, opus_data) do
#     :ok ->
#       :ok

#     {:error, reason} ->
#       IO.puts("Error sending opus data to ASR: #{reason}")
#       {:error, reason}
#   end
# end
# end

# @impl true
# def handle_cast({:run_tts,text}, state) do
#   # SockWorkstation.SockCoopClient.send_listening(state["self_pid"])
#   IO.puts("run_sent_tts")
#  SockWorkstation.SockCoopClient.send_speaking(CoopMgrGen.start_link(state.session_id).coop_pid)

#   {:ok, tts_pid} =
#     CoopHelperSup.start_tts_client({state.session_id,text})

#   # Process.send_after(state.self_pid, :tts_started, 250)

#   state =
#     state
#     |> Map.put(:tts_pid, tts_pid)
#     # |> Map.put(:tts_text, text)
#     # |> Map.put(:asr_starting, true)
#     # |> Map.put(:asr_task_id, asr_task_id)
#     # |> Map.put(:is_inited, false)

#   # |> Map.delete(:tts_task_id)
#   {:noreply, state}
# end

# def handle_cast({:asr_starting, pid}, state) do

#   {:noreply, %{state | asr_starting: true, bin_buffer: <<>>, chunk_num: 0, asr_pid: pid}}
# end

#   def handle_cast(:asr_stopping, state) do
#   {:noreply, %{state | asr_starting: false, asr_pid: nil}}
# end

# def handle_cast(:clear_buffer, state) do
#   {:noreply, %{state | bin_buffer: <<>>, chunk_num: 0}}
# end

# # 辅助函数：发送缓冲区数据到处理模块
# defp send_to_processor(session_id, opus_data) do
#   # 这里可以调用其他模块处理数据，例如编码或发送到网络
#   # IO.puts("Sending buffer data to processor: #{byte_size(opus_data)} bytes")
#   {:ok, pcm_data} = SockWorkstation.CodecNifGen.decode_opus(session_id, opus_data)
#   IO.puts("Sending buffer data to processor: #{byte_size(pcm_data)} bytes")
#   pcm_data
#   # SockWorkstation.PcmEncoderGen.pcm_to_opus(state.session_id, data)
# end

# 辅助函数：设置定时器
# defp schedule_send(interval, pid) do
#   Process.send_after(pid, :tts_started, interval)
# end

# def running_tts(session_id, asr_pid) do
#   GenServer.cast(via_tuple(session_id, "tts_helper"), {:tts_started, asr_pid})
# end

# def stopping_asr(session_id) do
#   GenServer.cast(via_tuple(session_id, "tts_helper"), :asr_stopping)
# end

# def start_sent_tts(session_id, text) do
#   GenServer.cast(via_tuple(session_id, "tts_helper"), {:run_tts, text})
# end

# def frsent_opus(session_id, opus_data) do
#   GenServer.cast(via_tuple(session_id, "tts_helper"), {:recv_opus, opus_data})
# end

# def push_data(session_id, data) do
#   GenServer.cast(via_tuple(session_id, "tts_helper"), {:opus_data, data})
# end

# def get_state(session_id) do
#   GenServer.call(via_tuple(session_id, "tts_helper"), :get_state)
# end

# def clear_buffer(session_id) do
#   GenServer.cast(via_tuple(session_id, "tts_helper"), :clear_buffer)
# end

# if byte_size(pcm_chunk) == @chunk_size do

# SockWorkstation.TTSHelperGen.TTSMembrane.send_to_processor(
# {:pcm_chunk, state.session_id, state.coop_pid, pcm_chunk},%{})
# {:ok, opus_data}=SockWorkstation.CodecNifGen.encode_pcm(state.session_id, pcm_chunk)
#  SockWorkstation.SockCoopClient.send_binary(state.coop_pid, opus_data)
# end

# stream = File.stream!("lvguang.pcm", [:binary], @chunk_size)
# stream = File.stream!("by.16000.pcm", [:binary], @chunk_size)
# Process.send_after(state.self_pid, :test_opus, div(File.stat!("lvguang.pcm").size, 32000))
# Enum.each(stream, fn chunk ->
#   if byte_size(chunk) == @chunk_size do
#     {:ok, opus_data} = SockWorkstation.CodecNifGen.encode_pcm(state.session_id, chunk)

#     # Task.start(fn ->
#     SockCoopClient.send_binary(state.coop_pid, opus_data)
#     Process.sleep(48)
#     # end)
#     #  else
#     # SockWorkstation.CoopMgrGen.run_tts_result(state.session_id)
#   end
# end)

# {:error, _reason} ->
#   {:noreply, state}
# end

# with {:ok, bin} <-
#        SockWorkstation.PanCouchbeamClient.get_stream_attachment(
#          "aod_opus_1",
#          "song_by_16000",
#          "by.16000.pcm"
#        ) do
#   send(state.tts_helper_pid, {:runfetch_mscpcm, bin})
# end

# stream = File.stream!("llh.pcm", [:binary], @chunk_size)
# # stream = File.stream!("by.16000.pcm", [:binary], @chunk_size)
# # Process.send_after(state.self_pid, :test_opus, div(File.stat!("llh.pcm").size, 32000))
# Enum.each(stream, fn chunk ->
#   if byte_size(chunk) == @chunk_size do
#     {:ok, opus_data} = SockWorkstation.CodecNifGen.encode_pcm(state.session_id, chunk)

#     # Task.start(fn ->
#     SockCoopClient.send_binary(state.coop_pid, opus_data)
#     Process.sleep(54)
#     # end)
# #  else
#   # SockWorkstation.CoopMgrGen.run_tts_result(state.session_id)
#   end
# end)
# {:noreply, state}
